package com.qyer.commons.test.concurrent;

import static com.mongodb.client.model.Projections.include;
import static com.qyer.commons.utils.DBUtils.tryConnect;
import static org.apache.commons.dbutils.DbUtils.closeQuietly;

import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.Filters;
import com.qyer.commons.concurrent.ArrayListBulk;
import com.qyer.commons.concurrent.PoisonPill;
import com.qyer.commons.concurrent.assemblingline.AbstractHandler;
import com.qyer.commons.concurrent.assemblingline.AssemblerGroup;
import com.qyer.commons.concurrent.assemblingline.AssemblingLine;
import com.qyer.commons.concurrent.assemblingline.HandleException;
import com.qyer.commons.concurrent.assemblingline.ResultWriter;
import com.qyer.commons.concurrent.assemblingline.StatisticContext;
import com.qyer.commons.mc.converter.Converter;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.commons.lang3.builder.ToStringStyle;
import org.bson.Document;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Collections;
import java.util.List;
import java.util.Map;

/**
 * User: Z J Wu Date: 2016/06/02 Time: 17:30 Package: com.qyer.commons.concurrent
 */
public class CDBMember implements PoisonPill<CDBMember>, Comparable<CDBMember> {

  public static class String2CDBMemberConverter extends Converter<String, CDBMember> {

    @Override
    public CDBMember convert(String s) throws Exception {
      if (StringUtils.isBlank(s)) {
        return null;
      }
      return CDBMember.createBasicCDBMember(Integer.parseInt(s));
    }
  }

  public static class EmailQueryHandler extends AbstractHandler<ArrayListBulk<CDBMember>,
    ArrayListBulk<CDBMember>> {

    private int bulkSize;

    private final String loadEmailSql;

    public EmailQueryHandler(StatisticContext context, int bulkSize) {
      super(context);
      this.bulkSize = bulkSize;
      StringBuilder sb = new StringBuilder();
      sb.append("SELECT t.uid, t.email FROM newbbs_utf8.cdb_members AS t WHERE t.uid in (");
      sb.append('?');
      for (int i = 0; i < this.bulkSize - 1; i++) {
        sb.append(", ?");
      }
      sb.append(") AND t.email IS NOT NULL AND t.email <> ''");
      this.loadEmailSql = sb.toString();
    }

    @Override
    public ArrayListBulk<CDBMember> handle(ArrayListBulk<CDBMember> bulk) throws HandleException {
      if (bulk == null) {
        return null;
      }
      Connection conn = null;
      PreparedStatement pstmt = null;
      ResultSet rs = null;
      List<CDBMember> list = bulk.getList();
      Collections.sort(list);
      Map<Integer, String> map = Maps.newHashMapWithExpectedSize(bulk.size());
      try {
        conn = tryConnect(TestMainDBResourceManager.getInstance().getDataSource(), 10,
                          "EmailQueryHandler");
        pstmt = conn.prepareStatement(loadEmailSql);
        for (int i = 0; i < list.size(); i++) {
          pstmt.setInt(i + 1, list.get(i).getId());
        }
        for (int i = list.size() + 1; i <= bulkSize; i++) {
          pstmt.setInt(i, 0);
        }
        rs = pstmt.executeQuery();
        while (rs.next()) {
          map.put(rs.getInt("uid"), rs.getString("email"));
        }
      } catch (Exception e) {
        throw new HandleException(e);
      } finally {
        closeQuietly(conn, pstmt, rs);
      }
      for (CDBMember cdbMember : list) {
        cdbMember.setEmail(map.get(cdbMember.getId()));
      }
      return bulk;
    }
  }

  public static class DeviceIdQueryHandler extends AbstractHandler<ArrayListBulk<CDBMember>,
    ArrayListBulk<CDBMember>> {

    public DeviceIdQueryHandler(StatisticContext context) {
      super(context);
    }

    @Override
    public ArrayListBulk<CDBMember> handle(ArrayListBulk<CDBMember> bulk) throws HandleException {
      if (bulk == null) {
        return null;
      }
      List<Integer> uidList = Lists.newArrayListWithExpectedSize(bulk.size());
      for (CDBMember cdbMember : bulk.getList()) {
        uidList.add(cdbMember.getId());
      }
      MongoDatabase database = TestMainMongoDBResourceManager.getInstance().getMongoClient()
                                                             .getDatabase("dc");
      MongoCollection<Document> collection = database.getCollection("uid_doraid_1");
      MongoCursor<Document> cur = collection.find(Filters.in("_id", uidList))
                                            .projection(include("_id", "devices")).iterator();
      Map<Integer, List<String>> resultMap = Maps.newHashMapWithExpectedSize(bulk.size());
      Document document;
      while (cur.hasNext()) {
        document = cur.next();
        try {
          resultMap.put(document.getInteger("_id"), (List<String>) document.get("devices"));
        } catch (Exception e) {
          e.printStackTrace();
        }
      }
      List<String> devices;
      for (CDBMember cdbMember : bulk.getList()) {
        devices = resultMap.get(cdbMember.getId());
        cdbMember.setDevices(devices);
      }
      return bulk;
    }
  }

  @JsonProperty("uid")
  private int id;
  @JsonProperty("email")
  private String email;
  @JsonProperty("devices")
  private List<String> devices;

  private CDBMember() {
    this.id = 0;
  }

  private CDBMember(int id) {
    this.id = id;
  }

  public static CDBMember createBasicCDBMember(int id) {
    if (id <= 0) {
      throw new IllegalArgumentException("id cannot <= 0");
    }
    return new CDBMember(id);
  }

  @Override
  @JsonIgnore
  public boolean isPill() {
    return this.id == 0;
  }

  @Override
  public CDBMember createPill() {
    return new CDBMember();
  }

  public int getId() {
    return id;
  }

  public void setId(int id) {
    this.id = id;
  }

  public String getEmail() {
    return email;
  }

  public void setEmail(String email) {
    this.email = email;
  }

  public List<String> getDevices() {
    return devices;
  }

  public void setDevices(List<String> devices) {
    this.devices = devices;
  }

  @Override
  public String toString() {
    ToStringBuilder tsb = new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE);
    if (isPill()) {
      tsb.append("(PILL)");
    } else {
      tsb.append("id", id).append("email", email).append("device", devices);
    }
    return tsb.toString();
  }

  @Override
  public int compareTo(CDBMember o) {
    if (o == null) {
      return 1;
    }
    return this.id - o.getId();
  }

  public static void main(String[] args) throws Exception {
    System
      .setProperty("test.db.main.properties.location", "/Users/WuZijing/dsp/db.main.properties");
    System
      .setProperty("test.db.mongo.properties.location", "/Users/WuZijing/dsp/db.mongo.properties");

    String in = "/Users/WuZijing/securecrt_rzsz/query_result.csv", out = in + ".result";
    CDBMember.String2CDBMemberConverter converter = new CDBMember.String2CDBMemberConverter();

    StatisticContext context = new StatisticContext();
    int emailBulkSize = 128, deviceBulkSize = 128;
    CDBMember.EmailQueryHandler emailQuerier = new CDBMember.EmailQueryHandler(context,
                                                                               emailBulkSize);
    CDBMember.DeviceIdQueryHandler deviceIdQuerier = new CDBMember.DeviceIdQueryHandler(context);
    ResultWriter resultWriter = new ResultWriter(context, out);

    AssemblingLine<CDBMember> builder = new AssemblingLine<>(context);
    builder.setFileDataGenerator("FileDataGenerator", in, converter);
    builder.setFileSummarizer("FileOutputSummarizer", resultWriter);
    builder.addAssemblerGroup("EmailAssembler", 20, emailQuerier, emailBulkSize)
           .addAssemblerGroup("DeviceAssembler", 20, deviceIdQuerier, deviceBulkSize);
    builder.connect();

    System.out.println(builder.getRawDataGenerator());
    for (AssemblerGroup<CDBMember> group : builder.getAssemblerGroups()) {
      System.out.println(group);
    }
    System.out.println(builder.getSummarizer());
    builder.start();
  }
}
